package distributed;

import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;

import common.*;

public class Node extends Thread{
	private int clkspd_ms;
	private long localtime_s;
	private int drift;
	private double smoothing = 1.0;

	private long ip;
	private int pid;
	
	private Middleware network;
	
	private int counter;
	private int synced_before_s = 0;
	private int logical_counter;
	
	public ConcurrentLinkedQueue<Message> msg_q;	// a message box (public)
	
	private int msg_id = 0;	
	private int sync_timeout_s;
	private boolean sync_begin = false;
	
	private ArrayList<NodeRecord> nodes;
	private ArrayList<Long> node_time;			// local queue (private) to store the time of other nodes
	private ArrayList<Integer> receive_interval;
	
	private boolean test_nodes_alive = false;
	private int interval;
	
	private boolean sleep_raised = false;
	private int sleep_left_s;
	private int sleep_counter;
	
	public Node( Middleware newNetwork ){
		// hw
		clkspd_ms = Config.TICK_NORMAL_MS;
		drift = ( new Random() ).nextInt( 3 * clkspd_ms / 2 ) - clkspd_ms / 2;	// drift = [ -clkspd_ms/2, 2*clkspd_ms ] 
		// sw
		counter = Config.MAX_COUNTER;
		pid = ( new Random() ).nextInt( Config.MAX_PID );
		logical_counter = 0;
		
		// msg q
		msg_q = new ConcurrentLinkedQueue<Message>();
		
		// network
		network = newNetwork;
		ip = network.connect( pid, this );
		
		nodes = new ArrayList<NodeRecord>();
		node_time = new ArrayList<Long>();
		receive_interval = new ArrayList<Integer>();
		
		print_log( "instantiated and joined to the network." );
	}
	
	/**
	 * Run() - Thread body.
	 * 		It periodically generates ticks and update the node timer.
	 * 		It also handles messages received and will be sent.
	 */
	
	public void run(){
		while( true ){
			
			try{
				// for preemptive sleeping
				if( sleep_raised ){
					sleep( clkspd_ms );		// don't account drift
					
					if( --sleep_counter == 0 ){
						sleep_counter = Config.MAX_COUNTER;
						print_log( "sleeping" );
						
						if( --sleep_left_s == 0 ){
							sleep_raised = false;
							int sleep_time =  (int)(Config.ROOT_SLEEP_TIMEOUT_S * clkspd_ms /(( clkspd_ms + drift) * smoothing));
							localtime_s = localtime_s + sleep_time;
							
							smoothing = 1.0;
							print_log( "... and I'm back!" );
						}
					}
					
					msg_q.poll();	// emptying msg_q: ignore all incoming messages while sleeping
					
					continue;
				}
				
				// clock (tick)
				sleep((int)(( clkspd_ms + drift) * smoothing));
			}
			catch( InterruptedException e ){
				e.printStackTrace();
			}
			
			// timer
			if( --counter == 0 ){				
				// timer interrupt
				localtime_s++;
				synced_before_s++;
				
				// counter reset
				counter = Config.MAX_COUNTER;
				
				// resync after every RESYNC_PERIOD_S seconds
				if(synced_before_s >= Config.RESYNC_PERIOD_S ){
					//send SYNC_REQ
					send( ++msg_id, Def.SYNC_REQ, 0, 0L, "" + localtime_s, logical_counter);
					print_log( "has broadcasted a recync signal." );
					sync_begin = true;
					sync_timeout_s = -1;
					nodes = new ArrayList<NodeRecord>(network.routing_table.keySet());
					
					// timer reset
					synced_before_s = 0;
					logical_counter++;
				}
				
				if(sync_begin){
					boolean adjustment_raised = false;
					long avg_time_s = localtime_s;
					
					sync_timeout_s++;
					
					// received all time from other nodes
					if(node_time.size() == network.routing_table.size()-1)
					{
						long total_time_s = 0;
						for(int i = 0; i < node_time.size(); i++){
							// current_time = sum( other_node_time + net_delay + queueing_delay )
							long other_node_time = node_time.get(i);
							int net_delay = receive_interval.get(i)/2;
							int queueing_delay = sync_timeout_s - receive_interval.get(i);
							
							long current_node_time = other_node_time + net_delay + queueing_delay;
							total_time_s = total_time_s + current_node_time;
						}
						total_time_s = total_time_s + localtime_s;
						avg_time_s = total_time_s/(node_time.size()+1);
						node_time.clear();
						receive_interval.clear();
						
						// time adjustment
						adjustment_raised = true;	// - CharmGil Added
						//localtime_s = avg_time_s;
						//print_log("my time has been sychronized.");
					
						//timer reset
						sync_begin = false;
					}
					else
					{
						if(sync_timeout_s >= Config.RESYNC_TIMEOUT_S){
							if(!node_time.isEmpty())
							{
								long total_time_s = 0;
								for(int i = 0; i < node_time.size(); i++){
									// current_time = sum( other_node_time + net_delay + queueing_delay )
									long current_node_time = node_time.get(i) - receive_interval.get(i)/2 + sync_timeout_s;
									total_time_s = total_time_s + current_node_time;
								}
								total_time_s = total_time_s + localtime_s;
								avg_time_s = total_time_s/(node_time.size()+1);
								node_time.clear();
								receive_interval.clear();
								
								// time adjustment
								adjustment_raised = true;	// - CharmGil Added
								//localtime_s = avg_time_s;
								//print_log("my time has been sychronized.");
								
								if(!nodes.isEmpty())
								{
									test_nodes_alive = true;
									interval = Config.REQUEST_TIMEOUT_S + 1;
									
									Iterator<NodeRecord> itr = nodes.iterator();
									while(itr.hasNext())
									{
										NodeRecord rcv = itr.next();
										Node dest = network.routing_table.get(rcv);
										
										send(msg_id, Def.REQ, dest.get_pid(), dest.get_ip(), "alive?", logical_counter);
										logical_counter++;
									}
								}
								
								//timer reset
								sync_begin = false;
							}//end-of-if(!node_time.isEmpty())
						}//end-of-if(sync_timeout_s)
					}//end-of-if(node_time.size())-else
					
					// Smooth Adjustment - CharmGil Added
					if( adjustment_raised ){
						long diff = localtime_s - avg_time_s;
						
						if( diff >= Config.RESYNC_PERIOD_S / ( double )Config.SMOOTHING_RATE ){
							// if diff is larger than sync period_s, put the process on hold for Config.RESYNC_PERIOD_S - 1 sec
							smoothing = Config.MAX_COUNTER * ( Config.RESYNC_PERIOD_S - 1 ) / ( double )( clkspd_ms + drift );
						}else{
							smoothing = 1.0 - diff * Config.SMOOTHING_RATE / ( double )Config.RESYNC_PERIOD_S;
							smoothing /= ( 1.0 + diff / ( double )synced_before_s );
							smoothing = 1 / smoothing;
							smoothing = ( smoothing <= 0 ) ? Config.RESYNC_PERIOD_S / ( double )( clkspd_ms + drift ) : smoothing;	// prevent backward adjustment
							if( smoothing == Double.POSITIVE_INFINITY )	smoothing = 1.0;
							if( smoothing == Double.NaN )	smoothing = 1.0;
							
							if( -diff >= Config.RESYNC_PERIOD_S / ( double )Config.SMOOTHING_RATE ){
								double extra = -diff / ( Config.SMOOTHING_RATE / ( double )Config.RESYNC_PERIOD_S );
								smoothing = smoothing / extra; 
							}
								
							//smoothing = ( smoothing == Double.POSITIVE_INFINITY ) ? 1.0 : smoothing;	// prevent permanent sleep
							//smoothing = ( smoothing == Double.NaN ) ? 1.0 : smoothing;	// prevent permanent sleep
						}
						
						print_log( "\tdrft=" + String.format( "%3d", drift ) + "    diff=" + String.format( "%3d", diff )
								//+ "\t s-d=" + df2.format( ( synced_before_s - diff ) )
								+ "   =>   smoothing= " + String.format( "%3.2f", smoothing )
								+ "\ttick= "  + String.format( "%3.2f", ( clkspd_ms + drift ) * smoothing )
								+ "\tsize(msg_q):" + msg_q.size()
								+ "\tsize(node_time):" + node_time.size() );
						
						//synced_before_s = 0;
						//send( Def.ACK, incoming_msg.getSpid(), incoming_msg.getSip(), "resc" + incoming_msg.getMid(), logical_counter );
						
						adjustment_raised = false;
					}
					
				}//end-of-if(sync_begin)
				
				if(test_nodes_alive == true)
				{
					interval--;
					if(interval == 0)
					{
						if(!nodes.isEmpty())
						{
							send(msg_id, Def.REQ, -1, -1L, "dead", logical_counter);
							nodes.clear();
						}
						test_nodes_alive = false;
					}
				}
			}
			
			Message incoming_msg = msg_q.poll();
			if( incoming_msg != null ){
				// logical counter
				logical_counter = Math.max( logical_counter + 1, incoming_msg.getMid() + 1 );
				
				// msg proc
				switch( incoming_msg.getHeader() ){
					case Def.RESYNC:
						if(incoming_msg.getMid() == msg_id){
							if(incoming_msg.getContent()!= null)
								
							{
								if( incoming_msg.getContent().substring( 0, 4 ).equals( "sync" ) ){
									node_time.add(Long.parseLong(incoming_msg.getContent().substring( 4 )));
									receive_interval.add(0);
									nodes.remove(new NodeRecord(incoming_msg.getSpid(),incoming_msg.getSip()));
									send(msg_id, Def.ACK, incoming_msg.getSpid(), incoming_msg.getSip(), "resc" + incoming_msg.getMid(), logical_counter);
								}
							}
							// to acknowledge
						}
						break;
						
					case Def.SYNC_REQ:
						send(incoming_msg.getMid(), Def.RESYNC, incoming_msg.getSpid(), incoming_msg.getSip(), "sync" + localtime_s, logical_counter);
						break;
						
					case Def.REQ:
						if(incoming_msg.getContent().equals("alive?"))
						{
							send(incoming_msg.getMid(), Def.ACK, incoming_msg.getSpid(), incoming_msg.getSip(), "yes", logical_counter);
						}
						break;
						
					case Def.ACK:
						// received the acknowledgment
						if(incoming_msg.getMid() == msg_id){
							if(incoming_msg.getContent()!=null)
							{
								if(incoming_msg.getContent().equals("yes")){
									nodes.remove(new NodeRecord(incoming_msg.getSpid(),incoming_msg.getSip()));
									send(msg_id, Def.ACK, incoming_msg.getSpid(), incoming_msg.getSip(), "OK", logical_counter);
								}
							}
						}
						break;
						
					case Def.MW_REQ:
						if(incoming_msg.getContent().equals("alive?"))
						{
							send(incoming_msg.getMid(), Def.ACK, -1, -1L, "yes", logical_counter);
						}
						break;
						
					case Def.MW_NOTIFY:
						break;
						
					case Def.NAK:	// CharmGil
						break;
						
					case Def.ROOT_KILL:
						network.disconnect( ip, pid );
						print_log( "I hate you guys! (Terminated)" );
						try{
							this.stop();
						}catch( Exception e ){}
						// end-of-thread: no more execution
						break;
					
					case Def.ROOT_SLEEP:
						sleep_left_s = Config.ROOT_SLEEP_TIMEOUT_S;
						sleep_counter = Config.MAX_COUNTER;
						sleep_raised = true;
						break;
						
					default:
						print_log( "received an unkown msg from " + incoming_msg.getSpid()  + "|"
								+ Utils.ipToString( incoming_msg.getRip() ) + ".", Def.STDERR );
					}
				}
			}
	}
	
	public int send( int msg_id, int msg_header, int rcv_pid, Long rcv_ip, String appx, int msg_counter )
	{
//		print_log( "**msg** " + msg_header );
		
		Message outgoing_msg = new Message( msg_id, msg_header, rcv_pid, rcv_ip, pid, ip );
		if( appx != null ){
			outgoing_msg.setContent( appx );
		}
		
		// send msg
		if( rcv_pid == 0 ){	// 0 to broadcast
			Collection<Node> all_nodes = network.routing_table.values();
			Iterator<Node> itr = all_nodes.iterator();
			int br_count = 0;	// #msg broadcasted
			while( itr.hasNext() ){
				Node br_rcv = itr.next();	// br_rcv = broadcast, receiver
				
				if( br_rcv.get_pid() != pid || ( br_rcv.get_pid() == pid && br_rcv.get_ip() != ip ) ){
						br_rcv.msg_q.offer( outgoing_msg );
						//print_log( "WR_msg(" + outgoing_msg.getHeader() + ")*: " + br_rcv.msg_q.offer( outgoing_msg ) );
						br_count++;
				}
			}
			print_log( "br_msg(" + msg_header + ") is sent to " + br_count + " nodes in the network (" + outgoing_msg.getMid() + ")." );		
			
			return br_count;
			
		}
		else{	// 1-to-1 comm
			if(rcv_pid == -1)
			{
				if(appx.equals("dead")){
					Iterator<NodeRecord> itr = nodes.iterator();
					while (itr.hasNext())
					{
						NodeRecord dead_node = itr.next();
						appx = appx + dead_node.toString() + ";";
					}
					outgoing_msg.setContent(appx);
					network.msg_q.offer(outgoing_msg);					
				}
			}
			else{
				Node receiver = network.routing_table.get( new NodeRecord( rcv_pid, rcv_ip ));
				receiver.msg_q.offer( outgoing_msg );
			}			
			return 1;
		}
	}
	
	public void recv(){
		
	}
	
	public long get_ip(){
		return ip;
	}
	
	public int get_pid(){
		return pid;
	}
	
	public long get_localtime_s(){
		return localtime_s;
	}
		
	public long get_localtime_ms(){
		return localtime_s * 1000;
	}
	
	public void set_localtime_ms( long new_localtime_ms ){
		localtime_s = new_localtime_ms / 1000;
	}
	
	public void set_localtime_s( long new_localtime_s ){
		localtime_s = new_localtime_s;
	}
	
	public void print_log( String log ){
		print_log( log, Def.STDOUT );
	}
	
	public void print_log( String log, byte target ){
		if( !Config.VERBOSE )
			return;
		
		if( target == Def.STDOUT ){
			System.out.print( "[MSG|" + pid + ":" + Utils.ipToString( ip ) + "(" + localtime_s + ")] " );
			System.out.println( log );
		}else if( target == Def.STDERR ){
			System.err.print( "[ERR|" + pid + ":" + Utils.ipToString( ip ) + "(" + localtime_s + ")] " );
			System.err.println( log );
		}
	}
	
	public String toString(){
		//return "Node(" + pid + ":" + ip + ")";
		return "Node(" + pid + ":" + Utils.ipToString( ip ) + ")";
	}
}
